package com.imooc.producer.b_order.partition;

import com.utils.Const;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.List;

/**
 * @author 折戟沉沙铁未销
 * @version V1.0
 * @date 2024/4/9-2024
 * @Description: 分区有序性。
 *      分区：指的是队列。  分区一词来源于kafka
 */
public class PartitionOrderlyProducer {
    public static void main(String[] args) throws Exception {
        //定义生产者
        DefaultMQProducer producer = new DefaultMQProducer("partition_orderly");
        producer.setNamesrvAddr(Const.NAME_SERVER_SINGLE_LOCAL);

        //定义topic主题对应的队列数量 （该参数可以不设置。主要是为了区分和全局有序性）
        //可以看到，这里并没像全局有序性那样，直接定义 queueNum=1
        //因为，分区有序性，只要保证每一批有序的消息发送到同一个队列中就可以。使用 MessageQueueSelector 接口就已经能够满足了。
        producer.setDefaultTopicQueueNums(4);

        //启动生产者
        producer.start();

        for (int i = 0; i < 10; i++) {
            Integer orderId = i;
            byte[] body = ("Hi, partition orderly" + i).getBytes();
            Message message = new Message("partition_orderly_topic", "tags_" + i, "key_" + i, body);
            /*
             * 保证消息的有序性，本质就是将需要保证顺序的这一批消息都发送到一个队列中。因为队列本身就是先进先出，队列是能够保证消息的顺序的。
             * 因此，需要保证顺序的消息，发送到一个队列中。这样这些消息在消费的时候就是顺序的。
             * 怎么保证同一批消息发送到一个队列中？
             *      MessageQueueSelector  这个接口给我我们一个方案。
             *      通过实现该接口，在该接口中可以根据我们自己的一个算法，指定消息发送哪个队列中。
             *      比如：有一批消息需要保证顺序，这批消息中的 parentId 属性是一样的，  那么我们就可以使用属性进行算法的运算，计算出一个队列。
             *      这批消息的 parentId 都是一样的，最终每个消息运算的结果都是一样的，这样这一批消息最终会选择到同一个队列进行发送。
             *      这样就可以实现消息的顺序。
             */
            producer.send(message, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    Integer id = (Integer) arg;
                    int num = id % mqs.size();
                    return mqs.get(num);
                }
            }, orderId);
        }

        //关闭生产者
        producer.shutdown();
    }
}